﻿using RockBus.DataContracts;
using RockBus.Logging.Interfaces;
using RockBus.Messages.Interfaces;
using RockBus.PerformanceCounters.Interfaces;
using RockBus.Routing.Interfaces;
using RockBus.ServiceModel;
using System;
using System.Reflection;
using System.ServiceModel;
using System.ServiceModel.Channels;
using System.Threading.Tasks;
using System.Transactions;

namespace RockBus.Routing
{
    public class ReceivePort : IReceivePort, IMessageProcessor
    {
        private const string OperationCounterName_ReceiveMessage = "ReceiveMessage";
        private const string OperationCounterName_PublishMessage = "PublishMessage";

        public ReceivePort(
            ILogger logger,
            IPerformanceCounterFactory performanceCounterFactory,
            IMessageFactory messageFactory,
            ISubscriptionConfigurationManager subscriptionConfigurationManager,
            IQueueManager queueManager,
            IPipelineManager pipelineManager)
        {
            this.Logger = logger;
            this.PerformanceCounterFactory = performanceCounterFactory;
            this.MessageFactory = messageFactory;
            this.SubscriptionConfigurationManager = subscriptionConfigurationManager;
            this.QueueManager = queueManager;
            this.PipelineManager = pipelineManager;

            this.SubscriptionConfigurationManager.OnSubscriptionConfigurationChanged += SubscriptionConfigurationManager_OnSubscriptionConfigurationChanged;
        }

        private ILogger Logger { get; set; }

        private IPerformanceCounterFactory PerformanceCounterFactory { get; set; }

        private IMessageFactory MessageFactory { get; set; }

        private ISubscriptionConfigurationManager SubscriptionConfigurationManager { get; set; }

        private IQueueManager QueueManager { get; set; }

        private IPipelineManager PipelineManager { get; set; }

        public PublisherConfiguration PublisherConfiguration { get; set; }

        public Task<Message> ProcessMessage(Message message, Transaction transaction, ExchangeTypeEnum exchangeType = ExchangeTypeEnum.OneWay)
        {
            try
            {
                var tcs = new TaskCompletionSource<Message>();

                this.CopyMessageContext(message);

                var remoteEndpoint = message.Properties.GetReceiveAddress();

                RoutingHeader routingHeader = null;
                CorrelationHeader correlationHeader = null;
                if (message.Headers.Action.Equals(StringConstants.PersistenceMessageAction))
                {
                    this.StartOperationCounter(tcs, OperationCounterName_PublishMessage);

                    var mcFromPersistence = this.MessageFactory.CreateMessageContextFromPersistenceMessage(message, transaction, tcs);
                    this.QueueManager.ProcessMessage(mcFromPersistence);
                    return tcs.Task;
                }

                // Extract any headers
                //  This should axctually no longer occur, as RoutingHeader and CorrelationHeader are now part of PersistenceMessage
                this.MessageFactory.ExtractHeaders(message, out routingHeader, out correlationHeader);

                var operationName = (routingHeader == null) ? OperationCounterName_ReceiveMessage : OperationCounterName_PublishMessage;
                this.StartOperationCounter(tcs, operationName);

                // Get the messageId or assign a new one
                var messageId = Guid.NewGuid();
                if (routingHeader != null)
                {
                    messageId = routingHeader.ProcessingData.MessageId;
                }
                else if (correlationHeader != null)
                {
                    messageId = correlationHeader.MessageId;
                }
                message.Properties.SetMessageId(messageId);

                //MessageData md = new MessageData(
                //    mc.ProcessingData.CorrelationId,
                //    message.Headers.Action,
                //    remoteEndpoint,
                //    OperationContext.Current.Channel.LocalAddress.Uri.AbsoluteUri,
                //    string.Empty,
                //    MessageEventEnum.Received,
                //    LogLevelEnum.Info);
                this.Logger.LogDebug(MethodBase.GetCurrentMethod(), "RECEIVED: Id: {0}; RemoteEndpoint: {1}; LocalEndpoint: {2}; MessageAction: {3}",
                    messageId,
                    remoteEndpoint,
                    OperationContext.Current.Channel.LocalAddress.Uri.AbsoluteUri,
                    message.Headers.Action
                    );

                var rpi = this.GetReceivePortInfo(message);
                if ((rpi != null) &&
                    (rpi.Detail != null) &&
                    (!string.IsNullOrEmpty(rpi.Detail.RequestPipelineId)))
                {
                    message = this.PipelineManager.ProcessInboundRequest(message, rpi.Detail);
                }

                var mc = this.MessageFactory.CreateRequestMessageContext(message, transaction, tcs, exchangeType, rpi, routingHeader, correlationHeader);

                this.QueueManager.ProcessMessage(mc);
                return tcs.Task;
            }
            catch (Exception exc)
            {
                this.Logger.LogError(MethodBase.GetCurrentMethod(), exc.ToString());
                throw;
            }
        }

        private void StartOperationCounter(TaskCompletionSource<Message> tcs, string operationName)
        {
            var oc = this.PerformanceCounterFactory.CreateOperationCounter(operationName);
            oc.Start();

            tcs.Task.ContinueWith((t) =>
            {
                oc.Complete();
            }, TaskContinuationOptions.OnlyOnRanToCompletion);
            tcs.Task.ContinueWith((t) =>
            {
                oc.Abort();
            }, TaskContinuationOptions.OnlyOnFaulted);
        }

        private void CopyMessageContext(Message message)
        {
            OperationContext.Current.CopyMessageContext();
            TopicHeader topicHeader = OperationContext.Current.IncomingMessageHeaders.FindHeader<TopicHeader>(TopicHeader.HeaderName, StringConstants.RockBusNamespace, true);
            if (null != topicHeader)
            {
                OperationContext.Current.IncomingMessageProperties.SetTopicHeader(topicHeader);
            }
        }

        public MessageProcessorEnum ProcessingStep { get { return MessageProcessorEnum.ReceivePort; } }

        public void ProcessMessage(IMessageContext messageContext)
        {
            this.Logger.LogInfo(MethodBase.GetCurrentMethod(), "Processing MessageContext {0}", messageContext.ToString());

            Message response = null;
            if (messageContext.ResponseMessage != null)
            {
                response = messageContext.ResponseMessage.Message;
            }

            if ((response != null) &&
                (messageContext.ReceivePortInfo != null) &&
                (!string.IsNullOrEmpty(messageContext.ReceivePortInfo.Detail.ResponsePipelineId)))
            {
                response = this.PipelineManager.ProcessOutboundResponse(messageContext.Message.Message, messageContext.ReceivePortInfo.Detail);
                messageContext = this.MessageFactory.CreateResponseMessageContext(messageContext, response);
            }

            messageContext.ProcessingData.MessageDirection = MessageDirectionEnum.End;
            this.QueueManager.ProcessMessage(messageContext);
        }

        private ReceivePortInfo GetReceivePortInfo(Message message)
        {
            // If a ReceivePortInfo needs to be applied, we expect the ReceivePortInfoId to be present in the MessageProperties
            //  This should be set by the ServiceEndpointBehavior when the endpoint was created.
            var publisherReceivePortInfo = message.Properties.GetPublisherReceivePortInfo() as PublisherReceivePortInfo;
            if (publisherReceivePortInfo == null)
            {
                return null;
            }

            ReceivePortInfo rpi = null;

            var pc = this.PublisherConfiguration;
            if ((pc == null) ||
                (!pc.TryFindReceivePortInfo(publisherReceivePortInfo, out rpi)) ||
                (rpi == null))
            {
                throw new ApplicationException(
                    string.Format("The PublisherConfiguration for PublisherId {0} and ReceivePortId {1} could not be determined. Verify the PublisherId and ReceivePortId configuration settings.",
                    publisherReceivePortInfo.PublisherId,
                    publisherReceivePortInfo.ReceivePortId));
            }

            return rpi;
        }

        private void SubscriptionConfigurationManager_OnSubscriptionConfigurationChanged(object sender, SubscriptionConfigurationChangedEventArgs e)
        {
            this.PublisherConfiguration = e.PublisherConfiguration;
        }
    }
}